package com.jacky.iot.handler.impl;

import com.jacky.iot.config.MqttServerConfig;
import com.jacky.iot.handler.MqttHandler;
import com.jacky.iot.manager.EndpointManager;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.vertx.mqtt.MqttEndpoint;
import io.vertx.mqtt.MqttTopicSubscription;
import io.vertx.mqtt.messages.MqttPublishMessage;
import io.vertx.mqtt.messages.MqttSubscribeMessage;
import io.vertx.mqtt.messages.MqttUnsubscribeMessage;

import java.util.Set;

public class Mqtt3Handler implements MqttHandler {
    private final EndpointManager endpointManager;
    private final MqttServerConfig config;

    public Mqtt3Handler(EndpointManager endpointManager, MqttServerConfig config) {
        this.endpointManager = endpointManager;
        this.config = config;
    }

    @Override
    public void handleConnection(MqttEndpoint endpoint) {
        // MQTT 3.1.1 的认证处理
        if (config.isEnableAuth()) {
            if (endpoint.auth() == null || !authenticate(endpoint)) {
                endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);
                return;
            }
        }

        endpointManager.addClient(endpoint.clientIdentifier(), endpoint);
        System.out.println("[MQTT 3] Client connected: " + endpoint.clientIdentifier());

        endpoint.disconnectHandler(v -> {
            endpointManager.removeClient(endpoint.clientIdentifier());
            System.out.println("[MQTT 3] Client disconnected: " + endpoint.clientIdentifier());
        });
    }

    @Override
    public void handlePublish(MqttEndpoint endpoint, MqttPublishMessage message) {
        System.out.printf("[MQTT 3] Received message on [%s]: %s%n",
                message.topicName(), message.payload().toString());
        endpointManager.publishToTopic(
                message.topicName(),
                message.payload(),
                message.qosLevel(),
                message.isDup(),
                message.isRetain()
        );
    }

    @Override
    public void handleSubscribe(MqttEndpoint endpoint, Set<MqttTopicSubscription> topics, MqttSubscribeMessage message) {
        topics.forEach(topic -> {
            endpointManager.addSubscription(endpoint.clientIdentifier(), topic);
            System.out.println("[MQTT 3] Client subscribed: " + endpoint.clientIdentifier() + " -> " + topic);
            // MQTT 3 需要明确回复 SUBACK
        });
        endpoint.subscribeAcknowledge(message.messageId(), topics.stream().map(MqttTopicSubscription::qualityOfService).toList());
    }

    @Override
    public void handleUnsubscribe(MqttEndpoint endpoint, Set<String> topics, MqttUnsubscribeMessage message) {
        topics.forEach(topic -> {
            endpointManager.removeSubscription(endpoint.clientIdentifier(), topic);
            System.out.println("[MQTT 3] Client unsubscribed: " + endpoint.clientIdentifier() + " -> " + topic);
            // MQTT 3 需要明确回复 UNSUBACK
            endpoint.unsubscribeAcknowledge(message.messageId());
        });
    }

    private boolean authenticate(MqttEndpoint endpoint) {
        return config.getUsers().containsKey(endpoint.auth().getUsername()) &&
                config.getUsers().get(endpoint.auth().getUsername()).equals(endpoint.auth().getPassword());
    }
}
